package com.yujiahao;


import org.apache.flink.kinesis.shaded.com.amazonaws.ClientConfiguration;
import org.apache.flink.kinesis.shaded.com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
import org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.AmazonKinesis;
import org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.AmazonKinesisClientBuilder;
import org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.model.*;
import org.junit.Before;
import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
import software.amazon.awssdk.services.kinesis.model.ListShardsRequest;
import software.amazon.awssdk.services.kinesis.model.ListShardsResponse;

import java.util.concurrent.TimeUnit;

import org.junit.Test;

import java.util.List;
import java.util.function.Consumer;

import static org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.model.ScalingType.UNIFORM_SCALING;

/**
 * @Description: 使用kinesis客户端进行流的创建
 * @Author: yujiahao
 * @Date: 2023/7/28 17:44
 * @Version: 1.0
 */

public class KinesisClient {


    private static AmazonKinesis client;
    private String streamName = "test-kds-api";
    /**
     * 创建客户端对象
     */
    @Before
    public void createClient() {
        //构建客户端对象
        AmazonKinesisClientBuilder clientBuilder = AmazonKinesisClientBuilder.standard();
        ClientConfiguration clientConfiguration = new ClientConfiguration();
        //DefaultAWSCredentialsProviderChain() 会按照以下顺序查找 AWS 凭证：环境变量、系统属性、或者配置文件中
        clientBuilder.setCredentials(new DefaultAWSCredentialsProviderChain());
        clientBuilder.setClientConfiguration(clientConfiguration);
        client = clientBuilder.build();

    }

    /**
     * 创建、更新、删除流
     */
    @Test
    public void createStream() {
        int shardCount = 8;

        //要创建流的名字
        String myStreamName = streamName;
        //创建请求
        CreateStreamRequest createStreamRequest = new CreateStreamRequest();
        createStreamRequest.setStreamName(myStreamName);
        //如果不设置就是按需模式的流，设置的话就是预置集群的流
        createStreamRequest.setShardCount(shardCount);
        //创建流,我电脑配置了客户端连接文件，所以直接就可以访问，不用设置访问凭证
//        client.createStream( createStreamRequest );
        //删除流（这个api比在控制台删除的快）
//        client.deleteStream(myStreamName);
        //更新分片
        client.updateShardCount(new UpdateShardCountRequest().withStreamName(myStreamName).withTargetShardCount(shardCount).withScalingType(UNIFORM_SCALING));

        DescribeStreamRequest describeStreamRequest = new DescribeStreamRequest();

        //Stream arn and stream name can't be empty at the same time
        describeStreamRequest.setStreamName(myStreamName);

    //如果流不存在，describeStream 将引发异常。因此，请将 describeStream 调用包括在 try/catch 块中。
        long startTime = System.currentTimeMillis();
        long endTime = startTime + (10 * 60 * 1000);
        while (System.currentTimeMillis() < endTime) {
            try {
                Thread.sleep(20 * 1000);
            } catch (Exception e) {
            }

            try {
                DescribeStreamResult describeStreamResponse = client.describeStream(describeStreamRequest);
                String streamStatus = describeStreamResponse.getStreamDescription().getStreamStatus();
                if (streamStatus.equals("ACTIVE")) {
                    break;
                }
                //
                // sleep for one second
                //
                try {
                    Thread.sleep(1000);
                } catch (Exception e) {
                }
            } catch (ResourceNotFoundException e) {
            }
        }
        if (System.currentTimeMillis() >= endTime) {
            throw new RuntimeException("Stream " + myStreamName + " never went active");
        }
    }

    /**
     * 列出流
     */
    @Test
    public  void listStreams() {

        //首先创建 ListStreamsRequest 的一个新实例
        ListStreamsRequest listStreamsRequest = new ListStreamsRequest();
        //调用应返回最多20个流,未指定值，KDA会返回小于或等于账户中数字的流数量
        listStreamsRequest.setLimit(20);
        ListStreamsResult listStreamsResult = client.listStreams(listStreamsRequest);
        //将返回的流名称存储在streamNames列表
        List<String> streamNames = listStreamsResult.getStreamNames();

        /**
         * getHasMoreStreams方法检查除了初始调用中返回的流之外是否还有其他可用流listStreams。
         * 如果是这样，代码将setExclusiveStartStreamName使用上次调用中返回的最后一个流的名称来调用该方法listStreams。
         * setExclusiveStartStreamName方法会导致下一次调用 listStreams在该流之后开始。
         * 然后，该调用返回的流名称组将添加到列表中streamNames。
         * 此过程持续进行，直到所有流名称都已收集到列表中。
         */

        while (listStreamsResult.getHasMoreStreams()) {
            if (streamNames.size() > 0) {
                listStreamsRequest.setExclusiveStartStreamName(streamNames.get(streamNames.size() - 1));
            }
            /**
             * 返回的流listStreams可以处于以下状态之一：可以使用describeStream查看
             * CREATING
             * ACTIVE
             * UPDATING
             * DELETING
             */
            listStreamsResult = client.listStreams(listStreamsRequest);
            streamNames.addAll(listStreamsResult.getStreamNames());
        }
        //[algo-data-stream-dev, algo-data-stream-prod, algo-data-stream-staging, app-point-dev, app-point-prod, app-solve-lambda-dev, app-solve-lambda-prod, app-solve-lambda-stage, app_log_dev, test-kds-api, test_bigdata_stream]
        System.out.println(streamNames);
    }

    /**
     * 列出分片,推荐使用新的 ListShards API
     */
    @Test
    public  void getShards(){

        //构建一个分片请求

        KinesisAsyncClient asyncClient = KinesisAsyncClient.builder().build();
        //注意导包的问题
        ListShardsRequest request = ListShardsRequest
                .builder().streamName(streamName)
                .build();
        //得到分片列表
        try {
            ListShardsResponse response = asyncClient.listShards(request).get(5000, TimeUnit.MILLISECONDS);
            System.out.println(response.toString());
        } catch (Exception e) {
            System.out.println(e.getMessage());
        }
    }

}

